package com.shujia.tf

import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.streaming.api.scala._

object Demo4KeyBy {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(2)

    val listDS: DataStream[String] = env.socketTextStream("master", 8888)

    val wordsDS: DataStream[String] = listDS.flatMap(_.split(","))

    /**
      * keyBy: 将相同的key发生到同一个task中，默认是hash分区
      *
      */
    val keyByDS: KeyedStream[String, String] = wordsDS.keyBy(word => word)

    //keyByDS.print()

    val javaApiDS: KeyedStream[String, String] = keyByDS.keyBy(new KeySelector[String, String] {
      override def getKey(value: String): String = {
        value
      }
    })

    javaApiDS.print()

    env.execute()

  }

}
